package com.campus.merchant.service;

import com.campus.common.message.MealReadyMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

/**
 * 出餐消息生产者
 * 负责向Kafka发送出餐消息
 */
@Service
@Slf4j
public class MealReadyMessageProducer {

    private static final String TOPIC_MEAL_READY = "meal-ready";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @Resource
    private ObjectMapper objectMapper;

    /**
     * 发送出餐消息
     * @param orderId 订单ID
     * @param deliveryId 配送员ID
     */
    public void sendMealReadyMessage(String orderId, Integer deliveryId) {
        try {
            MealReadyMessage message = new MealReadyMessage(orderId, deliveryId);
            String messageJson = objectMapper.writeValueAsString(message);
            
            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC_MEAL_READY, orderId, messageJson);
            
            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    log.info("出餐消息发送成功，订单号: {}, 配送员ID: {}, offset: {}",
                            orderId, deliveryId, result.getRecordMetadata().offset());
                }

                @Override
                public void onFailure(Throwable ex) {
                    log.error("出餐消息发送失败，订单号: {}, 配送员ID: {}", orderId, deliveryId, ex);
                }
            });
        } catch (Exception e) {
            log.error("发送出餐消息异常，订单号: {}, 配送员ID: {}", orderId, deliveryId, e);
        }
    }
}